package com.atguigu.app.dwd.log;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.util.DateFormatUtil;
import com.atguigu.util.KafkaUtil;
import com.google.inject.internal.cglib.core.$DefaultGeneratorStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author hefeihong
 * @version 1.0
 * @date 2022/9/3 11:44
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //设置状态后端
        /*env.enableCheckpointing(5*60*1000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setCheckpointTimeout(3*60*1000L);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");*/

        //TODO 3 读取kafka主题topic_log 中的数据
        String topicLog = "topic_log";
        String groupId = "base_log_app";
        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.getKafkaConsumer(topicLog, groupId));

        //TODO 4 清洗并转换数据为JSON格式，非JSON格式的剔除
        OutputTag<String> dirty = new OutputTag<String>("dirty"){};
        SingleOutputStreamOperator<JSONObject> process = streamSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(value);
                    out.collect(jsonObject);
                } catch (Exception e) {
                    e.printStackTrace();
                    ctx.output(dirty, value);
                }
            }
        });

        //TODO 5 根据mid 分组数据
        KeyedStream<JSONObject, String> keyedStream = process.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                JSONObject common = value.getJSONObject("common");
                return common.getString("mid");
            }
        });

        //TODO 6 根据键控状态（mid，dt）完成新旧访客的标记修复
        SingleOutputStreamOperator<JSONObject> map = keyedStream.map(new RichMapFunction<JSONObject, JSONObject>() {
            private ValueState<String> lastState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                lastState = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastState", String.class));
            }

            @Override
            public JSONObject map(JSONObject value) throws Exception {
                JSONObject common = value.getJSONObject("common");
                String is_new = common.getString("is_new");
                String lastDtState = lastState.value();
                String lastDt = DateFormatUtil.toDate(value.getLong("ts"));
                if ("1".equals(is_new)) {
                    //真的新访客 value为空或者value和当前登录的日期是同一天
                    if (lastDtState == null) {
                        lastState.update(lastDt);
                    } else if (!lastDtState.equals(lastDt)) {
                        //重新安装的app用户，重置为0
                        common.put("is_new", "0");
                    }
                } else {
                    //标记为0
                    if (lastDtState == null) {
                        //修改状态日期为前一天
                        String yesterdday = DateFormatUtil.toDate(value.getLong("ts") - 1000L * 60 * 60 * 24);
                        lastState.update(yesterdday);
                    }
                }
                return value;
            }
        });

        //TODO 7 将数据拆分为6种数据流
        //page页面当做主流 另外5种 start err action display appvideo作为侧输出流
        OutputTag<String> startOutputTag = new OutputTag<String>("start"){};
        OutputTag<String> errOutputTag = new OutputTag<String>("err"){};
        OutputTag<String> actionOutputTag = new OutputTag<String>("action"){};
        OutputTag<String> displayOutputTag = new OutputTag<String>("display"){};
        OutputTag<String> appVideoOutputTag = new OutputTag<String>("appVideo"){};

        SingleOutputStreamOperator<String> pageStream = map.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {
                //topic_log中有三种日志 start page appvideo
                //1错误数据输出到侧输出流
                String err = value.getString("err");
                if (err != null) {
                    ctx.output(errOutputTag, err);
                }
                value.remove("err");

                //启动日志输出到侧输出流
                String appVideo = value.getString("appVideo");
                String start = value.getString("start");
                if (start != null) {
                    ctx.output(startOutputTag, value.toJSONString());
                }else if (appVideo!=null){
                    ctx.output(appVideoOutputTag,value.toJSONString());
                }
                else {
                    //当前为页面日志
                    //获取公共信息
                    JSONObject common = value.getJSONObject("common");
                    Long ts = value.getLong("ts");
                    JSONObject page = value.getJSONObject("page");


                    //打散输出action数据
                    JSONArray actions = value.getJSONArray("actions");
                    if (actions != null) {
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject jsonObject = actions.getJSONObject(i);
                            jsonObject.put("common", common);
                            jsonObject.put("ts", ts);
                            jsonObject.put("page", page);
                            ctx.output(actionOutputTag, jsonObject.toJSONString());
                        }
                    }
                    value.remove("actions");

                    //打散输出display数据
                    JSONArray displays = value.getJSONArray("displays");
                    if (displays != null) {
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            display.put("common", common);
                            display.put("ts", ts);
                            display.put("page", page);
                            ctx.output(displayOutputTag, display.toJSONString());
                        }
                    }
                    value.remove("displays");

                    //主流输出page数据
                    out.collect(value.toJSONString());
                }
            }
        });

        pageStream.print("page>>>>>");

        DataStream<String> startStream = pageStream.getSideOutput(startOutputTag);
        DataStream<String> errStream = pageStream.getSideOutput(errOutputTag);
        DataStream<String> actionStream = pageStream.getSideOutput(actionOutputTag);
        DataStream<String> displayStream = pageStream.getSideOutput(displayOutputTag);
        DataStream<String> appVideoStream = pageStream.getSideOutput(appVideoOutputTag);

        startStream.print("start>>>>>>");
        errStream.print("err>>>>>");
        actionStream.print("action>>>>>");
        displayStream.print("display>>>>>");
        appVideoStream.print("appVideo>>>>>");

        //TODO  8 将5中数据流写回到对应的kafka主题
        // 定义不同日志输出到 Kafka 的主题名称
        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";
        String appVideo_topic = "dwd_traffic_appvideo_log";

        pageStream.addSink(KafkaUtil.getKafkaProducer(page_topic));
        startStream.addSink(KafkaUtil.getKafkaProducer(start_topic));
        displayStream.addSink(KafkaUtil.getKafkaProducer(display_topic));
        actionStream.addSink(KafkaUtil.getKafkaProducer(action_topic));
        errStream.addSink(KafkaUtil.getKafkaProducer(error_topic));
        appVideoStream.addSink(KafkaUtil.getKafkaProducer(appVideo_topic));

        //TODO 9 执行

        env.execute(groupId);
    }
}
